package sg20.task2;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;



import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class TaskTwoDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        Configuration conf = new Configuration();

        // 全局变量传递:universityLineCnt
        conf.set("universityLineCnt", String.valueOf(getUniversityLineCnt(conf, args[1])));

        //如果输出文件夹存在,则删除
        FileSystem fs = FileSystem.get(conf);
        if(fs.exists(new Path(args[2])))
            fs.delete(new Path(args[2]), true);

        //1.get job
        Job job = Job.getInstance(conf);
        job.setJobName("Join Course and University");

        //2.set jar path
        job.setJarByClass(TaskTwoDriver.class);

        //3.connect with class of map and reduce
        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, TaskTwoCourseMapper.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, TaskTwoUniversityMapper.class);
        job.setReducerClass(TaskTwoReducer.class);

        //4.set the type of K output of map
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        //5.set the type of KV output of reduce
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //6.set input and output path
        FileOutputFormat.setOutputPath(job, new Path(args[2]));

        //7.submit job
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    //This function is to get the number of lines of "university.tbl"
    public static int getUniversityLineCnt(Configuration configuration, String hdfsPath) {
        int res = 0;
        Path path = new Path(hdfsPath);

        // 定义一个字符串用来存储文件内容
        try {
            FileSystem fileSystem = path.getFileSystem(configuration);
            FSDataInputStream fsDataInputStream = fileSystem.open(path);
            BufferedReader br = new BufferedReader(new InputStreamReader(fsDataInputStream));
            while (br.readLine() != null) {
                res++;
            }
            br.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return res;
    }

}
